package com.rocketmq.example;

import java.util.Iterator;
import java.util.List;
import java.util.Set;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragelyByCircle;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

public class DefaultPropertyForConsumer {

    public static void main(String[] args) throws Exception {
        // 属性
        // 消费者组
        // DefaultMQPushConsumer defaultMQPushConsumer = new
        // DefaultMQPushConsumer(ExampleMQConfig.GROUP.getName());
        /**
         * 算法名称	含义
            AllocateMessageQueueAveragely	平均分配算法
            AllocateMessageQueueAveragelyByCircle	基于环形平均分配算法
            AllocateMachineRoomNearby	基于机房临近原则算法
            AllocateMessageQueueByMachineRoom	基于机房分配算法
            AllocateMessageQueueConsistentHash	基于一致性hash算法
            AllocateMessageQueueByConfig	基于配置分配算法
         */
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(ExampleMQConfig.GROUP.getName(), null, new AllocateMessageQueueAveragelyByCircle());
        // 设置 Namesrv 地址
        defaultMQPushConsumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());
        // 设置消费模式为集群模式，也是默认的消费模式
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        // 指定消费开始的偏移量（上次消费偏移量、最大偏移量、最小偏移量、启动时间戳） 开始消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 消费者最小线程数
        defaultMQPushConsumer.setConsumeThreadMin(20);
        // 消费者最大线程数
        defaultMQPushConsumer.setConsumeThreadMax(20);
        // 消费线程每次消费的数量
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
        // push 模式下的任务时间间隔（push 模式下也是基于不断轮询拉取的封装）
        defaultMQPushConsumer.setPullInterval(0);
        // push 模式下任务拉取的条数
        defaultMQPushConsumer.setPullBatchSize(32);
        // 消息重试次数: -1 其实代表 16次 ，超过这个次数被认定为死信
        defaultMQPushConsumer.setMaxReconsumeTimes(-1);
        // 消费超时时间（消息可能阻塞正在舒勇的线程的最大时间，以分钟为单位）
        defaultMQPushConsumer.setConsumeTimeout(15);

        // 获取消费者对主题分配了哪些消息队列
        Set<MessageQueue> fetchSubscribeMessageQueueSet = defaultMQPushConsumer
                .fetchSubscribeMessageQueues(ExampleMQConfig.TOPIC.getName());

        Iterator<MessageQueue> iterator = fetchSubscribeMessageQueueSet.iterator();
        while (iterator.hasNext()) {
            System.err.println(iterator.next().getQueueId());
        }

        // 订阅
        // 基于主题订阅，消息过滤使用表达式
        defaultMQPushConsumer.subscribe(ExampleMQConfig.TOPIC.getName(), "*");
        // 基于主题订阅，消息过滤使用SQL
        defaultMQPushConsumer.subscribe(ExampleMQConfig.TOPIC.getName(), MessageSelector.bySql(" a between 0 and 3 "));
        // 基于主题订阅，消息过滤使用表达式
        defaultMQPushConsumer.subscribe(ExampleMQConfig.TOPIC.getName(), MessageSelector.byTag("TagA|TagB"));
        // 取消消息订阅
        defaultMQPushConsumer.unsubscribe(ExampleMQConfig.TOPIC.getName());

        // 注册监听器
        // 注册并发监听器
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                msgs.forEach(msg -> {
                    System.err.println(msg.getBody());
                    System.err.println(msg.getMsgId());
                });
                // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 顺序监听器
        defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                msgs.forEach(msg -> {
                    System.err.println(msg.getQueueId());
                    System.err.println(msg.getBody());
                });
                // 一会再处理这个消息，并不是放在重试队列，放入重试队列会造成消息消费的无序性
                // return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                return ConsumeOrderlyStatus.SUCCESS;
            }

        });
    }

}
